/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.jr.rest.util;

import cn.ac.iie.di.ban.jr.rest.global.GlobalParas;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import org.apache.avro.generic.GenericRecord;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType.Struct;
import static cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType.Structs;

/**
 *
 * @author Black-Laptop
 */
public class SendData {

    private static final Logger LOG = LoggerFactory.getLogger(SendData.class);
    private String topic = null;
    public SendData(String topic) throws ConfigurationException {
        this.topic = topic;
    }

    public void sendMessage(String datatype, List<String> colNames, REFieldType[] types, List<Object[]> datas) throws RESessionException, REConnectionException {
        //生成连接，参数为rocketmq的nameserver串。可以多个nameserver，分号隔开。由具体业务决定
        LOG.info("topic: " + topic);
        REConnection conn = new REConnection(GlobalParas.CONSUMER_NAMESERVER);

        //生成builder，参数为topic，由具体业务决定
        RESendSessionBuilder builder = (RESendSessionBuilder) conn.getSendSessionBuilder(topic);
        //设置数据结构，参数为每一列的列名、列类型、是否可以为空。注意： 如果设置为不能为空，那么如果添加数据时没有添加该列数据，会报错。
        for (int i = 0; i < colNames.size(); i++) {
            builder.addColumn((String) colNames.get(i), types[i], true);
            HashMap<String, REFieldType> detailMap = new HashMap<>();
            switch (types[i]) {
                case Struct:
//                    detailMap2.put("v", REFieldType.String);
//                    detailMap2.put("w", REFieldType.Long);
                    Struct.setDetail((String) colNames.get(i), detailMap);
                case Structs:
                    LOG.debug("Structs name: " + colNames.get(i));
                    HashMap<String, REFieldType> detailMap2 = new HashMap<>();
//                    detailMap2.put("v", REFieldType.String);
//                    detailMap2.put("w", REFieldType.Long);
                    Structs.setDetail((String) colNames.get(i), detailMap2);
                    break;
            }
//            switch (types[i]) {
//                case Struct:
////                    for (Object key : ((Map)datas.get(0)[i]).keySet()) {
////                        detailMap.put((String)key, RocketMQUtils.GetVarType(((Map)datas.get(0)[i]).get(key)));
////                        LOG.debug("Key = " + key);
////                    }
////                    Struct.setDetail(ArrCol[i], detailMap);
//                    HashMap<String, REFieldType> detailMap2 = new HashMap<>();
//                    detailMap2.put("v", REFieldType.String);
//                    detailMap2.put("w", REFieldType.Long);
//                    Structs.setDetail(colNames.get(i), detailMap2);
//                    break;
//                case Structs:
////                    Map[] mapArray = (Map[]) datas.get(0)[i];
////                    for (int indexMap = 0; indexMap < mapArray.length; indexMap++) {
////                        for (Object key :mapArray[indexMap].keySet()) {
////                            detailMap.put((String) key, RocketMQUtils.GetVarType(mapArray[indexMap].get(key)));
////                            LOG.debug("Key = " + key);
////                        }
////                    }
//                    String[] newCol=colNames.toArray(new String[colNames.size()]);
////                    Structs.setDetail(newCol[i], detailMap);
//                    detailMap.put("v", REFieldType.String);
//                    detailMap.put("w", REFieldType.Long);
//                    Structs.setDetail(colNames.get(i), detailMap);
//                    break;
//                default:
//            }

        }
        LOG.debug("colName: " + Arrays.toString(colNames.toArray()) + "  type: " + Arrays.toString(types));

        /**
         * 这里是设置struct或者structs的格式。 首先需要addColumn
         * 其次需要在对应的类型里面setDetail，参数是：列名，子列类型map
         */
        RESendSession session = null;
        try {
            session = (RESendSession) builder.build();
            session.start();
            //开始发送。
            //这里，DocSchemaName和UserDesc非必须，具体什么时候需要，由具体业务决定。
            //DocSchemaName设置一次后不会清空，每次flush都是同一个，如果需要改变，需要再次设置。
            //for (int r = 0; r < 10; ++r) {

            //添加数据。set完所有列后add即可添加一条数据。多条数据打一包发送。
            //各种set方法可以通过session.set补全来查看。set方法和REFieldType各类对应。
            //注意：尽量不要一条数据发送一次，这样会带来较大的额外开销
            //注意2：本地缓存数据量有大小限制，如果超过500M，会报错。
            //注意3：如果要用set（index，value）的方式设置数据，index从1开始。
            LOG.debug("datas size: " + datas.size());
            LOG.debug("type size: " + types.length);

            for (Object[] data : datas) {
                LOG.info(datatype + " send data: " + Arrays.toString(data));
                for (int i = 0; i < types.length; i++) {
                    try {
                        LOG.debug("column: " + colNames.get(i) + " type:" + types[i] + " value: " + data[i]);
                        switch (types[i]) {
                            case String:
                                session.setString((String) colNames.get(i), String.valueOf(data[i]));
                                break;
                            case Int:
                                session.setInt((String) colNames.get(i), Integer.valueOf(data[i].toString()));
                                break;
                            case Long:
                                session.setLong((String) colNames.get(i), Long.valueOf(data[i].toString()));
                                break;
                            case Float:
                                session.setFloat((String) colNames.get(i), Float.valueOf(data[i].toString()));
                                break;
                            case Double:
                                session.setDouble((String) colNames.get(i), Double.valueOf(data[i].toString()));
                                break;
                            case Binary:
                                session.setBinary((String) colNames.get(i), data[i].toString().getBytes());
                                break;
                            case Doc:
                                session.setDoc((String) colNames.get(i), (GenericRecord) data[i]);
                                break;
                            case Docs:
                                session.setDocs((String) colNames.get(i), (GenericRecord[]) data[i]);
                                break;
                            case Binaries:
                                session.setBinaries((String) colNames.get(i), (byte[][]) data[i]);
                                break;
                            case Doubles:
                                session.setDoubles((String) colNames.get(i), (double[]) data[i]);
                                break;
                            case Floats:
                                session.setFloats((String) colNames.get(i), (float[]) data[i]);
                                break;
                            case Longs:
                                session.setLongs((String) colNames.get(i), (long[]) data[i]);
                                break;
                            case Ints:
                                session.setInts((String) colNames.get(i), (int[]) data[i]);
                                break;
                            case Strings:
                                session.setStrings((String) colNames.get(i), (String[]) data[i]);
                                break;
                            case Boolean:
                                session.setBoolean((String) colNames.get(i), (boolean) data[i]);
                                break;
                            case Booleans:
                                session.setBooleans((String) colNames.get(i), (boolean[]) data[i]);
                                break;
                            case Struct:
                                session.setStruct((String) colNames.get(i), (Map) data[i]);
                                break;
                            case Structs:
                                session.setStructs((String) colNames.get(i), (Map[]) data[i]);
                                break;
                            default:
                                LOG.error("There is not the type of " + types[i]);
                        }
                    } catch (Exception ex) {
                        LOG.error(ExceptionUtils.getFullStackTrace(ex));
                    }

                }
                session.add();
            }
            //添加完所有数据，flush即可发送。
            session.flush();
            LOG.info(datatype + " Send Message " + (datas.size()));
            Thread.sleep(1);
        } catch (Exception ex) {
            LOG.error(ExceptionUtils.getFullStackTrace(ex));
        } finally {
            //session结束
            session.shutdown();
        }
    }

}
